package sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Statement;

public class SinkToMysql_Province extends RichSinkFunction<ClassObject.ProvinceResult> {
    //声明连接和预编译语句
    Connection connection = null;
    PreparedStatement insertStmt = null;
    PreparedStatement updateStmt = null;
    PreparedStatement replaceStmt = null;
    Statement stmt=null;
    // open属于生命周期，避免在invoke()中来一条数据就创建一次数据库连接
    // open属于生命周期，避免在invoke()中来一条数据就创建一次数据库连接
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://192.168.68.140:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "root");
        insertStmt = connection.prepareStatement("insert IGNORE into detail (name, value) values (?, ?)");
//        updateStmt = connection.prepareStatement("update detail set value = ? where name =?");
//        stmt = connection.createStatement();
            replaceStmt = connection.prepareStatement("replace into detail (name, value) values (?, ?)");
    }

    // 每来一条数据，调用连接，执行sql
    @Override
    public void invoke(ClassObject.ProvinceResult value, Context context) throws Exception {

        // 直接执行更新语句，如果没有更新就执行插入
        for(String i:value.getPrankSals()){
            String[] array = i.split(",");
//            String sql = "UPDATE detail set value = "+Double.valueOf(array[1])+"where name like '"+array[0]+"%' ;";
//            int result = stmt.executeUpdate(sql);

//            System.out.println(array[0]);
            replaceStmt.setString(1, array[0]);
            replaceStmt.setDouble(2, Double.valueOf(array[1]));
            replaceStmt.execute();
//            if(updateStmt.getUpdateCount() == 0){
//                System.out.println(1);
//                insertStmt.setString(1, array[0]);
//                insertStmt.setDouble(2, Double.valueOf(array[1]));
//                insertStmt.execute();
//            }
        }
        replaceStmt.setString(1, "总金额");
        replaceStmt.setDouble(2, value.getTotalPrice());
        replaceStmt.execute();
//        if(updateStmt.getUpdateCount() == 0){
//            insertStmt.setString(1, "总金额");
//            insertStmt.setDouble(2, value.getTotalPrice());
//            insertStmt.execute();
//        }
        replaceStmt.setString(1, "订单数量");
        replaceStmt.setDouble(2, value.getNum());
        replaceStmt.execute();
//        if(updateStmt.getUpdateCount() == 0){
//            insertStmt.setString(1, "订单数量");
//            insertStmt.setDouble(2, value.getNum());
//            insertStmt.execute();
//        }
    }

//    @Override
//    public void close() throws Exception {
//        insertStmt.close();
//        updateStmt.close();
//        connection.close();
//    }
}
